package com.cango.student.webflux;

import org.junit.Test;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

public class WebFlux {

    public static void first() {
        // Flux和Mono提供了多种创建数据流的方法，just就是一种比较直接的声明数据流的方式，其参数就是数据元素。
        Flux.just(1, 2, 3, 4, 5, 6); // 一个Flux对象代表一个包含0..N个元素的响应式序列
        Mono.just(1); // 一个Mono对象代表一个包含零/一个（0..1）元素的结果

        // 基于数组，集合和Stream
        Integer[] array = new Integer[]{1, 2, 3, 4, 5, 6};
        Flux.fromArray(array);
        List<Integer> list = Arrays.asList(array);
        Flux.fromIterable(list);
        Stream<Integer> stream = list.stream();
        Flux.fromStream(stream);

        // 首先，错误信号和完成信号都是终止信号，二者不可能同时共存；
        // 如果没有发出任何一个元素值，而是直接发出完成/错误信号，表示这是一个空数据流；
        // 如果没有错误信号和完成信号，那么就是一个无限数据流。
        // 只有完成信号的空数据流
        Flux.just();
        Flux.empty();
        Mono.empty();
        Mono.justOrEmpty(Optional.empty());
        // 只有错误信号的数据流
        Flux.error(new Exception("some error"));
        Mono.error(new Exception("some error"));
    }

    public static void second() {
//        Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::print);
//        System.out.println();
//        Mono.just(1).subscribe(System.out::println);
//        // 订阅并触发数据流
//        subscribe();
//// 订阅并指定对正常数据元素如何处理
//        subscribe(Consumer<? super T> consumer);
//// 订阅并定义对正常数据元素和错误信号的处理
//        subscribe(Consumer<? super T> consumer,
//                Consumer<? super Throwable> errorConsumer);
//// 订阅并定义对正常数据元素、错误信号和完成信号的处理
//        subscribe(Consumer<? super T> consumer,
//                Consumer<? super Throwable> errorConsumer,
//                Runnable completeConsumer);
//// 订阅并定义对正常数据元素、错误信号和完成信号的处理，以及订阅发生时的处理逻辑
//        subscribe(Consumer<? super T> consumer,
//                Consumer<? super Throwable> errorConsumer,
//                Runnable completeConsumer,
//                Consumer<? super Subscription> subscriptionConsumer);

        Flux.just(1, 2, 3, 4, 5, 6).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!"));

        // 错误信号
        Mono.error(new Exception("some error")).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!")
        );
    }

    /**
     * 1. 响应式之道 - 3 快速上手 - 2 Reactor
     * <p>
     * 使用StepVerifier进行单测。
     */
    @Test
    public void testViaStepVerifier() {
        // 先了解一个基本的单元测试工具——StepVerifier
        StepVerifier.create(generateFluxFrom1To6())
                // expectNext用于测试下一个期望的数据元素
                .expectNext(1, 2, 3, 4, 5, 6)
                // expectComplete用于测试下一个元素是否为完成信号
                .expectComplete()
                .verify();
        StepVerifier.create(generateMonoWithError())
                // expectErrorMessage用于校验下一个元素是否为错误信号
                .expectErrorMessage("some error")
                .verify();
    }

    private Flux<Integer> generateFluxFrom1To6() {
        return Flux.just(1, 2, 3, 4, 5, 6);
    }

    private Mono<Integer> generateMonoWithError() {
        return Mono.error(new Exception("some error"));
    }


    /**
     * 1. 响应式之道 - 3 快速上手 - 2 Reactor
     * <p>
     * 测试基本的Operator。
     */
    @Test
    public void testSimpleOperators() throws InterruptedException {
        // map + filter
        StepVerifier.create(Flux.range(1, 6)
                .filter(i -> i % 2 == 1)
                .map(i -> i * i))
                .expectNext(1, 9, 25)
                // verifyComplete()相当于expectComplete().verify()。
                .verifyComplete();

        // flatMap
        StepVerifier.create(
                Flux.just("flux", "mono")
                        .flatMap(s -> Flux.fromArray(s.split("\\s*")) // 对于每一个字符串s，将其拆分为包含一个字符的字符串流；
                                .delayElements(Duration.ofMillis(1000))) // 对每个元素延迟100ms
                        // 注doOnNext方法是“偷窥式”的方法，不会消费数据流
                        .doOnNext(System.out::print))
                // 验证是否发出了8个元素。
                .expectNextCount(8)
                .verifyComplete();

        System.out.println();

        // zip - 一对一合并
        // Flux的zip方法接受Flux或Mono为参数，Mono的zip方法只能接受Mono类型的参数。
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Flux.zip(
                getZipDescFlux(),
                // 使用Flux.interval声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度；
                Flux.interval(Duration.ofMillis(200)))
//                .doOnComplete(countDownLatch::countDown)
                // zip之后的流中元素类型为Tuple2，使用getT1方法拿到字符串流的元素；定义完成信号的处理为countDown
                .subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown);
//

//        getZipDescFlux().zipWith(Flux.interval(Duration.ofMillis(200))).subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown);

        countDownLatch.await(10, TimeUnit.SECONDS);

        Flux.just("flux", "mono").count().subscribe(System.out::println);

    }

    private Flux<String> getZipDescFlux() {
        String desc = "Zip two sources together, that is to say wait   for all the sources to emit one element and combine these elements once into a Tuple2.";
        return Flux.fromArray(desc.split("\\s+"));
    }

    /**
     * 1. 响应式之道 - 3 快速上手 - 2 Reactor
     * <p>
     * 将同步阻塞调用转化为异步。
     */
    @Test
    public void testSyncToAsync() throws InterruptedException {
//        Schedulers.immediate(); // 当前线程
//        Schedulers.single(); // 可重用的单线程 这个方法对所有调用者都提供同一个线程来使用， 直到该调度器被废弃。如果你想使用独占的线程，请使用Schedulers.newSingle()
//        Schedulers.elastic(); // 弹性线程池 它根据需要创建一个线程池，重用空闲线程。线程池如果空闲时间过长 （默认为 60s）就会被废弃。对于 I/O 阻塞的场景比较适用。Schedulers.elastic()能够方便地给一个阻塞 的任务分配它自己的线程，从而不会妨碍其他任务和资源；
//        Schedulers.parallel(); // 固定大小线程池，所创建线程池的大小与CPU个数等同；
//        Schedulers.fromExecutorService(ExecutorService); // 自定义线程池（Schedulers.fromExecutorService(ExecutorService)）基于自定义的ExecutorService创建 Scheduler（虽然不太建议，不过你也可以使用Executor来创建）。
        // Flux和Mono的调度操作符subscribeOn和publishOn支持work-stealing。

        CountDownLatch countDownLatch = new CountDownLatch(1);


        Mono.fromCallable(() -> getStringSync()) // 使用fromCallable声明一个基于Callable的Mono；
                // 使用subscribeOn将任务调度到Schedulers内置的弹性线程池执行，弹性线程池会为Callable的执行任务分配一个单独的线程。
                .subscribeOn(Schedulers.elastic())
                .subscribe(System.out::println, null, countDownLatch::countDown);
        countDownLatch.await(10, TimeUnit.SECONDS);
    }

    private String getStringSync() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "Hello, Reactor!";
    }

    @Test
    public void testErrorHandling() {
        Flux.range(1, 6)
                .map(i -> 10 / (i - 3)) // 1
                .map(i -> i * i)
                .subscribe(System.out::println, System.err::println);


// result
//        25
//        100
//        java.lang.ArithmeticException: / by zero    //注：这一行是红色，表示标准错误输出
    }

    @Test
    public void testOnErrorReturn() {
        Flux.range(1, 6)
                .map(i -> 10 / (i - 3)) // 1
                .onErrorReturn(0) // 当发生异常时提供一个缺省值0,并跳出
                .map(i -> i * i)
                .subscribe(System.out::println, System.err::println);
    }

    @Test
    public void testOnErrorResume() {
        Flux.range(1, 6)
                .map(i -> 10 / (i - 3))
                .onErrorResume(e -> Mono.just(new Random().nextInt(6))) // 提供新的数据流
                .map(i -> i * i)
                .subscribe(System.out::println, System.err::println);
    }

    @Test
    public void testOnError() {
        try {
            Flux.range(1, 6)
                    .map(i -> 10 / (i - 3))
                    .onErrorMap(e -> new Exception(e)) // 生成exception
                    .map(i -> i * i)
                    .subscribe(System.out::println, e -> {
                        Exception ex = (Exception) e;
//                        throw e;
                    });
        } catch (Exception ex) {
            System.out.println("catch");
        }

    }

    @Test
    public void testOnError2() {

        try {
            Flux.range(1, 6)
                    .map(i -> 10 / (i - 3))
                    .doOnError(e -> {   // 1
                        System.out.println("uh oh, falling back, service failed for key :" + e.getMessage());    // 2
                    })
//                    .onErrorResume(e -> Mono.just(new Random().nextInt(6))) // oncomplete
                    .map(i -> i * i)
                    .doFinally(type -> {
                        if (type == SignalType.ON_ERROR)
                            System.out.println("ON_ERROR");
                    })
                    .subscribe(System.out::println, e -> {
                        Exception ex = (Exception) e;
//                        throw e;
                    });
        } catch (Exception ex) {
            System.out.println("catch");
        }

    }

    @Test
    public void testOnFinally() {

        Flux<Integer> flux = Flux.range(1, 6)
                .map(i -> i * i)
                .doFinally(type -> {
                    if (type == SignalType.ON_ERROR)
                        System.out.println("ON_ERROR");
                    if (type == SignalType.ON_COMPLETE)
                        System.out.println("ON_COMPLETE");
                }).take(2);
        // 只有订阅后，才能doFinally
        flux.subscribe(System.out::println, System.out::println);
    }

    @Test
    public void testRetry() {

        try {
            Flux.range(1, 6)
                    .map(i -> 10 / (3 - i))
                    .retry(1)
                    .subscribe(System.out::println, System.err::println);
            Thread.sleep(100);  // 确保序列执行完
        } catch (Exception ex) {
            System.out.println("catch");
        }
    }

    @Test
    public void testBackpressure() {
        Flux.range(1, 6)    // Flux.range是一个快的Publisher；
                .doOnRequest(n -> System.out.println("Request " + n + " values..."))    // 在每次request的时候打印request个数；
                .subscribe(new BaseSubscriber<Integer>() {  // 通过重写BaseSubscriber的方法来自定义Subscriber；
                    @Override
                    protected void hookOnSubscribe(Subscription subscription) { // hookOnSubscribe定义在订阅的时候执行的操作；
                        System.out.println("Subscribed and make a request...");
                        request(1); // 订阅时首先向上游请求1个元素；
                    }

                    @Override
                    protected void hookOnNext(Integer value) {  // hookOnNext定义每次在收到一个元素的时候的操作；
                        try {
                            TimeUnit.SECONDS.sleep(1);  // sleep 1秒钟来模拟慢的Subscriber；
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("Get value [" + value + "]");    // 打印收到的元素；
                        request(1); // 每次处理完1个元素后再请求1个。
                    }
                });
    }

}
